package com.tc.vms.producer;

import com.google.protobuf.ByteString;
import com.rabbitmq.client.*;
import com.tc.vms.MessageWrap;
import com.tc.vms.connection.VmsConnectionManager;
import com.tc.vms.encoding.Message.KeyPair;
import com.tc.vms.encoding.Message.VMSMessage;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.TransactionMessageListener;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.message.SendResultType;
import com.tc.vms.utils.ServiceThread;
import com.tc.vms.utils.Utils;
import org.slf4j.Logger;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Created by yonghua.zhang on 2015/12/14.
 */
public class VmsProducerRmqMultiQueueImpl extends AbstractPublisher {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    protected final static ThreadLocal<AMQP.BasicProperties.Builder> vmsAmqpBuilderLocal = new ThreadLocal<AMQP.BasicProperties.Builder>() {

        @Override
        protected AMQP.BasicProperties.Builder initialValue() {
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties().builder();
            builder.contentEncoding("protobuf");
            builder.contentType("application/octet-stream");
            return builder;
        }
    };
    private final ConnectionFactory factory;
    private volatile Connection connection;
    private volatile Channel channel;
    private final String rmqExchangeName;
    private final Object sendLock = new Object();
    private final ServiceThread procShutdownSignal;
    private volatile boolean hasShutdownSignal;

    private final List<String> subOverlayQueues;
    private volatile String curShardsId;   //be used to deciding which connetion would be created.
    private final Map<Integer, String> delayExchangeCache = new HashMap<Integer, String>(16);

    public VmsProducerRmqMultiQueueImpl(VmsProducerProxy vmsProducerProxy, VmsConnectionManager vmsConnectionManager) {
        super(vmsProducerProxy, vmsConnectionManager);
        ChannelMetaData channelMetaData = vmsProducerProxy.getVmsMetaData();
        rmqExchangeName = channelMetaData.getRealChlNmOnBroker();
        factory = new ConnectionFactory();

        List<String> subQueues = channelMetaData.getChannelSubQueues();
        if (subQueues.size() > 1) { //the channel be consumed by multiQueues, get the overlay shards to send msg
            subOverlayQueues = getOverlayQueueShards(subQueues);
            if (subOverlayQueues.isEmpty()) {//general speaking, the thing couldn't happen.
                subOverlayQueues
                        .addAll(vmsConnectionManager.getVmsRouterManager().getQueueShardsInfo(subQueues.get(0)));
            }
        } else if (subQueues.size() == 1) { //only be consumed by one queue
            subOverlayQueues = vmsConnectionManager.getVmsRouterManager().getQueueShardsInfo(subQueues.get(0));
        } else {
            subOverlayQueues = Collections.emptyList();
        }

        procShutdownSignal = new ServiceThread() {
            @Override
            public String getServiceName() {
                return "Process-producer-shutdown-signal-thread_" + System.currentTimeMillis();
            }

            @Override
            public void run() {
                while (true) {
                    try {
                        if (hasShutdownSignal) {
                            TimeUnit.MILLISECONDS.sleep(3000);
                            LOGGER.error(
                                    "[VmsProducerRmqMultiQueueImpl] Rabbitmq channel exception, and recovery channel. curThreadId = [{}]",
                                    Thread.currentThread().getName());
                            synchronized (VmsProducerRmqMultiQueueImpl.this) {
                                if (isOpen) {
                                    while (channel == null || !channel.isOpen()) {
                                        try {
                                            innerShutdown();
                                            LOGGER.info(
                                                    "[VmsProducerRmqMultiQueueImpl] RestoreShutdown - innerShutdown operation successfully.");
                                            createChannel();
                                            LOGGER.info(
                                                    "[VmsProducerRmqMultiQueueImpl] RestoreShutdown - createChannel operation successfully.");
                                            hasShutdownSignal = false;
                                            break;
                                        } catch (Exception e) {
                                            TimeUnit.MILLISECONDS.sleep(3000);
                                            LOGGER.error(
                                                    "[VmsProducerRmqMultiQueueImpl] Rabbitmq channel exception, Recovery channel failed and retry after 3s interval.", e);
                                        }
                                    }
                                }
                            }
                        } else {
                            TimeUnit.MILLISECONDS.sleep(3000);
                        }
                    } catch (InterruptedException e) {
                        LOGGER.warn("Process-producer-shutdown-signal-thread shutdown.");
                        break;
                    }
                }
            }
        };
    }

    private final static AMQP.BasicProperties buildBasicProps(MessageWrap message) {
        AMQP.BasicProperties.Builder builder = vmsAmqpBuilderLocal.get();
        /* ChannelMetaData channelMetaData = vmsProducerProxy.getVmsMetaData();
           builder.userId(channelMetaData.getUserName())
                .appId(channelMetaData.getBelongAttr().getBelongAppname());*/
        /*        builder.timestamp(new Date(message.getBornTime()));
        builder.messageId(message.getMessageId());*/

        builder.deliveryMode(message.durable() ? 2 : 1);
        if (message.getTtl() > 0) {
            builder.expiration(String.valueOf(message.getTtl()));
        } else {
            builder.expiration(null);
        }
        if (message.getPriority() > 0) {
            builder.priority(message.getPriority());
        } else {
            builder.priority(null);
        }
        return builder.build();
    }

    public final static VMSMessage buildVmsMessage(MessageWrap message) {
        VMSMessage.Builder vmsBuilder = vmsBuilderLocal.get();
        vmsBuilder.clear();
        vmsBuilder.setPayload(ByteString.copyFrom(message.getPayload()));
        vmsBuilder.setMessageId(message.getMessageId());
        vmsBuilder.addAllRoutingKeys(message.getRoutingKeys());

  /*      if (message.getTtl() > 0) {
            vmsBuilder.setTtl(message.getTtl());
        }
        if (message.getPriority() > 0) {
            vmsBuilder.setWeight(message.getPriority());
        }*/
        KeyPair.Builder builder = vmsKeyBuilderLocal.get();
        for (Map.Entry<String, Object> withEntry : message.getProperties()) {
            builder.clear();
            builder.setKey(withEntry.getKey());
            builder.setValue(String.valueOf(withEntry.getValue()));
            vmsBuilder.addProperties(builder.build());
        }
        return vmsBuilder.build();
    }

    private final boolean processResponse(final Channel channel) throws TimeoutException, InterruptedException {
        switch (serviceSLA.sendMode()) {
            case SYNC:
                int waitForTimeout = serviceSLA.confirmTimeoutInMs();
                if (waitForTimeout == -1) {
                    channel.waitForConfirms();
                } else {
                    channel.waitForConfirms(waitForTimeout);
                }
                break;
            case ASYNC: //use batch send mode by async
                break;
            case ONEWAY:
            default:
                break;
        }
        return true;
    }


/*    private void asyncSend(Message[] messages) {

    }*/

    @Override
    public SendResult sendMessage(MessageWrap message) throws VmsClientException {
        SendResult sr = null;
        if (message.getRoutingKeys().size() > 0) {
            for (String routeKey : message.getRoutingKeys()) {
                sr = sendOneMessage(message, routeKey);
                //任何一个routekey发送出错 退出发送
                if (sr.getSendResultType() != SendResultType.VMS_SEND_OK) {
                    break;
                }
            }
        } else {
            throw new VmsClientException("Send rabbitmq message, need route-key attribute.",
                    VmsClientErrorCode.RABBITMQ_ERR_CODE_NO_ROUTE_KEY);
        }
        return sr;
    }

    private SendResult sendOneMessage(MessageWrap message, String routeKey) {
        int retryTimes = 0;
        SendResult sr = null;
        String shardsRouteKey = routeKey;
        if (curShardsId != null && !curShardsId.isEmpty()) {
            shardsRouteKey = Utils.joinShardsTags(routeKey, curShardsId);
        }
        AMQP.BasicProperties rmqProp = buildBasicProps(message);
        VMSMessage vmsMessage = buildVmsMessage(message);
        while ((sr == null || sr.getSendResultType() != SendResultType.VMS_SEND_OK) && retryTimes <= serviceSLA
                .retryTimesWhenSend()) {
            try {
                synchronized (sendLock) {
                    channel.basicPublish(rmqExchangeName, shardsRouteKey, message.mandatory(), false, rmqProp,
                            vmsMessage.toByteArray());
                    processResponse(channel);
                }
                sr = new SendResult(SendResultType.VMS_SEND_OK, message.getMessageId(), null);
                break;
            } catch (Exception e) {
                sr = new SendResult(SendResultType.VMS_SEND_FAILED, message.getMessageId(), e.toString());
                if (!serviceSLA.isFailFast()) {
                    Utils.sleep3s();
                }
            }
            retryTimes++;
        }
        return sr;
    }

    @Override
    public SendResult sendMessageByOrder(MessageWrap message, String partKey) throws VmsClientException {
        return sendMessage(message);
    }

    @Override
    public SendResult sendTransactionMessage(MessageWrap message, TransactionMessageListener listener)
            throws VmsClientException {
        throw new VmsClientException(
                "Send rabbitmq message failed. Not support the function of sendTransactionMessage ");
    }

    private ArrayList<String> getOverlayQueueShards(List<String> subQueues) {
        //if exists overlay, random get someone from overlay shards.
        //if no overlay shards, random get one shards to create connection, maybe exists performance issue.
        HashMap<String, Integer> shardsTagsStat = new HashMap<String, Integer>();
        ArrayList<String> tmpShards = new ArrayList<String>();
        int maxNum = 0;
        try {
            for (String tmpSubQueue : subQueues) {
                for (String tag : vmsConnectionManager.getVmsRouterManager().getQueueShardsInfo(tmpSubQueue)) {
                    if (shardsTagsStat.containsKey(tag)) {
                        shardsTagsStat.put(tag, shardsTagsStat.get(tag) + 1);
                    } else {
                        shardsTagsStat.put(tag, 1);
                    }
                    maxNum = Math.max(maxNum, shardsTagsStat.get(tag));
                }
            }

            if (maxNum > 1) {
                for (Map.Entry<String, Integer> entry : shardsTagsStat.entrySet()) {
                    if (entry.getValue().equals(maxNum)) {
                        tmpShards.add(entry.getKey());
                    }
                }
            }
            if (!tmpShards.isEmpty()) {
                Collections.shuffle(tmpShards);
            }
        } catch (Exception e) {
            LOGGER.warn("getOverlayQueueShards error.", e);
        }
        return tmpShards;
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        try {
            ChannelMetaData channelMetaData = vmsProducerProxy.getVmsMetaData();
            String vhost = channelMetaData.getVhost();
            String userName = channelMetaData.getUserName();
            String passwd = channelMetaData.getUserPasswd();
            //factory.setAutomaticRecoveryEnabled(true);
            factory.setRequestedHeartbeat(Utils.RMQ_HEART_BEAT_INTERVAL);
            factory.setVirtualHost(vhost.length() > 1 ? vhost.replace("/", "") : vhost);
            factory.setUsername(userName);
            factory.setPassword(passwd);
            factory.setConnectionTimeout(serviceSLA.confirmTimeoutInMs());
            return true;
        } catch (Exception e) {
            throw new VmsClientException(e.toString(),
                    e);
        }
    }

    private synchronized void createChannel() throws VmsClientException {
        if (channel == null) {
            try {
                if (connection == null || !connection.isOpen()) {
                    ChannelMetaData channelMetaData = vmsProducerProxy.getVmsMetaData();
                    if (!channelMetaData.isSharding()) {//no shards situation
                        Address[] addresses = Utils.shuffle(channelMetaData.getBrokerAddress());
                        connection = factory.newConnection(addresses);
                    } else { //some queue exists shards
                        Exception lastException = null;
                        boolean createdOk = false;

                        if (!this.subOverlayQueues.isEmpty()) {
                            Collections.shuffle(subOverlayQueues);
                            for (String shardAddr : subOverlayQueues) {
                                try {
                                    int brokerIdx = Utils.getBrokerIdByShardingTag(shardAddr) % channelMetaData
                                            .getBrokerAddress().length;
                                    Address[] tmpHostAddrs = new Address[]{
                                            channelMetaData.getBrokerAddress()[brokerIdx]};
                                    connection = factory.newConnection(tmpHostAddrs);
                                    curShardsId = shardAddr;
                                    createdOk = true;
                                    break;
                                } catch (Exception e) {
                                    lastException = e;
                                }
                            }
                            if (!createdOk) {
                                throw (lastException != null) ?
                                        new VmsClientException("failed to connect." + lastException, lastException) :
                                        new VmsClientException("failed to connect.");
                            }
                        } else {
                            Address[] addresses = Utils.shuffle(channelMetaData.getBrokerAddress());
                            connection = factory.newConnection(addresses);
                        }
                    }
                }
                channel = connection.createChannel();

                channel.addShutdownListener(new ShutdownListener() {
                    private volatile boolean hasShutdownNotify = false;

                    @Override
                    public void shutdownCompleted(ShutdownSignalException cause) {
                        if (!hasShutdownSignal) {
                            hasShutdownSignal = true;
                            LOGGER.warn("Rabbitmq channel has been shutdown by accident.");
                        }
                    }
                });
                if (serviceSLA.msgMode().equals(MsgMode.TRANSACT_MESG)) {
                    channel.txSelect();
                } else if (serviceSLA.sendMode().equals(SendMode.SYNC)) {
                    channel.confirmSelect();
                }

            } catch (Exception e) {
                innerShutdown();
                LOGGER.error("Create Rabbitmq channel failed.", e);
                throw new VmsClientException("Create Rabbitmq channel failed.", e);
            }
        } else if (!channel.isOpen()) {
            innerShutdown();
            throw new VmsClientException("Create Rabbitmq channel failed.");
        }
    }

    @Override
    public synchronized boolean start() throws VmsClientException {
        if (!isOpen) {
            isOpen = true;
            init(null);
            createChannel();
            //declareDelayExchanges();
            procShutdownSignal.start();
        }
        return isOpen;
    }

    @Override
    public synchronized void shutdown() {
        if (isOpen) {
            isOpen = false;
            hasShutdownSignal = false;
            procShutdownSignal.shutdown(true);
            innerShutdown();
        }
    }

    private void declareDelayExchanges() throws VmsClientException {
        if (channel != null && channel.isOpen()) {
            try {
                for (int iLvl = 1; iLvl <= Utils.MAX_DELAY_LEVEL; iLvl++) {
                    String delayExchangeNm = iLvl + ".vms.delay.exchange";
                    delayExchangeCache.put(iLvl, delayExchangeNm);
                    channel.exchangeDeclare(delayExchangeNm, "fanout", true, false, false, null);
                }
            } catch (Exception e) {
                throw new VmsClientException("declareDelayQueues failed.", e);
            }
        }
    }

    private synchronized void innerShutdown() {
        if (channel != null) {
            try {
                channel.close();
            } catch (Exception e) {
            }
            channel = null;
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (Exception e) {
            }
            connection = null;
        }
    }
}
